package com.gugee.ins.data.common.consumer;

import com.gugee.ins.data.common.msg.InsMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;
import java.util.stream.Collectors;

/**
 * Created with IntelliJ IDEA.
 * Description:抽象消费逻辑的入口
 * User: luolifeng
 * Date: 2020-12-24 16:28
 */
@Slf4j
public abstract class AbstractGugeeConsumerHandler<T> {

    public   void  handle(String who, List<T> messages, long acceptMsgMaxLimitTimes, Acknowledgment ack) {
        long start=System.currentTimeMillis();
        long acceptMsgTimeMillis= acceptMsgMaxLimitTimes*60*1000;
        String error="success";
        boolean expired=false;
        try {
            long isRange=start-acceptMsgTimeMillis;
            if(((InsMessage)messages.get(0)).getEventTime() < isRange ){
                expired=true;
                log.info("{},接收到消息数量：{},不在处理时限内消息数量：{},时限分钟：{}" , who,messages.size(),messages.size(),acceptMsgMaxLimitTimes);
            }else{
                dealMsg(messages);
            }

            ack.acknowledge();
        }catch (Exception e){
            e.printStackTrace();
            log.error("{},发生异常：{}" ,who,e.getMessage());
            error=e.getMessage();
        }finally {
            if(!expired) {
                long times = (System.currentTimeMillis() - start) / 1000;
                if (times < 1) {
                    times = 1;
                }
                if(!who.equals("InsMonitorEventListener")) {
                    log.info("{},消息数量: {},处理耗时: {}s,每秒处理消息数={}/s,结果：{}", who, messages.size(), times, messages.size() / times, error);
                }
            }
        }
    }
    public abstract void dealMsg(List<T> messages);
}
